package com.techsun.source;

import com.alibaba.fastjson.JSONObject;
import com.techsun.utils.MysqlUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.AbstractSource;
import org.apache.log4j.Logger;

import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;


/**
 * metric 各参数含义
 * OpenConnectionCount（打开的连接数）、
 * Type（组件类型）、
 * AppendBatchAcceptedCount（追加到channel中的批数量）、
 * AppendBatchReceivedCount（source端刚刚追加的批数量）、
 * EventAcceptedCount（成功放入channel的event数量）、
 * AppendReceivedCount（source追加目前收到的数量）、
 * StartTime（组件开始时间）、
 * StopTime（组件停止时间）、
 * EventReceivedCount（source端成功收到的event数量）、
 * AppendAcceptedCount（放入channel的event数量）
 */
public class MySqlSource extends AbstractSource implements Configurable, PollableSource {
    Logger logger = Logger.getLogger(MySqlSource.class);

    private String sourceUrl;
    private String checkPointUrl;
    private Long pollIntervalSecond;
    // metric 的统计类
    private SourceCounter sourceCounter;
    // 查询sql字符串, 设为私有变量, 防止重复创建
    private String selectSql;
    // checkpoint中间过滤表需要定期删除过期的活动数据, 此处做个标识, 用于每天执行一遍删除语句
    private String deleteExpiredActiveFlag = "";
    // 需要删除几天前的过期数据
    private Long deleteExpiredActiveIntervalDay;

    /**
     * 初始化checkpoint的数据库信息, 程序启动时执行一次
     */
    private void initTable() {
        MysqlUtil mysqlUtil = null;
        try {
            mysqlUtil = new MysqlUtil(checkPointUrl);
            mysqlUtil.execute("CREATE DATABASE IF NOT EXISTS `ma_checkpoint` /*!40100 DEFAULT CHARACTER SET utf8mb4 */; ");
            mysqlUtil.execute("CREATE TABLE IF NOT EXISTS `ma_checkpoint`.`tsr_marketing_activity_checkpoint` (" +
                    "  `id` varchar(46) NOT NULL COMMENT '营销活动表id'," +
                    "  `end_time` datetime DEFAULT NULL COMMENT '活动结束时间'," +
                    "  `period_flag` varchar(46) NOT NULL DEFAULT '' COMMENT '周期活动次数标识'," +
                    "  `checkpoint_status` int unsigned DEFAULT '1' COMMENT '该数据使用情况 0:未使用/1:已使用'," +
                    "  PRIMARY KEY (`id`,`period_flag`) " +
                    ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
        } catch (Exception e) {
            logger.error("init mysql table has error: ", e);
        } finally {
            if (mysqlUtil != null)
                mysqlUtil.close();
        }
    }

    @Override
    public synchronized void start() {
        this.initTable();
        this.sourceCounter.start();
        super.start();
    }

    @Override
    public void configure(Context context) {
        this.sourceUrl = context.getString("sourceUrl");
        this.checkPointUrl = context.getString("checkPointUrl", this.sourceUrl);
        this.pollIntervalSecond = context.getLong("pollIntervalSecond", 60L);
        this.deleteExpiredActiveIntervalDay = context.getLong("deleteExpiredActiveIntervalDay", 7L);
        String sourceDatabase = context.getString("sourceDatabase");
        String sourceTable = context.getString("sourceTable");
        String tableName = String.format("%s.%s", sourceDatabase, sourceTable);
        this.selectSql = "SELECT * FROM ( " +
                "SELECT id, end_time , CASE  " +
                "WHEN activity_type = 1 AND period_unit = 100000003 THEN date_format(curdate(), '%dD') " +
                "WHEN activity_type = 1 AND period_unit = 100000002 AND period_value = date_format(curdate(), '%w') THEN date_format(curdate(), '%uU%wW') " +
                "WHEN activity_type = 1 AND period_unit = 100000001 AND period_value = date_format(curdate(), '%d') THEN date_format(curdate(), '%mM%dD') " +
                "WHEN activity_type = 0 THEN '' " +
                "ELSE 'no_match' " +
                "END AS period_flag " +
                "FROM " + tableName +
                " WHERE audit_status = 100000003 AND status = 100000002 AND TIMESTAMPDIFF(MINUTE, current_timestamp(), begin_time) < 0 AND TIMESTAMPDIFF(MINUTE, current_timestamp(), end_time) > 0 " +
                ") t WHERE t.period_flag != 'no_match'";

        if (this.sourceCounter == null) {
            this.sourceCounter = new SourceCounter(this.getName());
        }
    }

    @Override
    public Status process() {
        // 查询数据
        MysqlUtil sourceMysql = null;
        String filterSql = "";
        try {
            sourceMysql = new MysqlUtil(sourceUrl);
            List<JSONObject> select = sourceMysql.select(this.selectSql);
            StringBuilder unionSql = new StringBuilder();
            boolean fieldFlag = true;
            for (JSONObject jsonObject : select) {
                if (fieldFlag) {
                    unionSql.append(String.format("select '%s' as id,'%s' as end_time,'%s' as period_flag union all ", jsonObject.getString("id"), jsonObject.getString("end_time"), jsonObject.getString("period_flag")));
                    fieldFlag = false;
                } else {
                    unionSql.append(String.format("select '%s','%s','%s' union all ", jsonObject.getString("id"), jsonObject.getString("end_time"), jsonObject.getString("period_flag")));
                }
            }
            if (unionSql.length() == 0) {
                Thread.sleep(this.pollIntervalSecond * 1000);
                return Status.BACKOFF;
            }
            // 去掉最后一个union all
            String unionSqlSubStr = unionSql.substring(0, unionSql.length() - 10);
            filterSql = String.format("SELECT t.id,t.end_time,t.period_flag FROM (%s) t LEFT JOIN ma_checkpoint.tsr_marketing_activity_checkpoint tsr ON t.id = tsr.id and t.period_flag = tsr.period_flag WHERE tsr.id IS NULL UNION select id,end_time,period_flag from ma_checkpoint.tsr_marketing_activity_checkpoint where checkpoint_status = 0 and TIMESTAMPDIFF(MINUTE, current_timestamp(), end_time) >= 0", unionSqlSubStr);
        } catch (Exception e) {
            logger.error("get resource mysql data has error: ", e);
            return Status.BACKOFF;
        } finally {
            if (sourceMysql != null)
                sourceMysql.close();
        }

        // 过滤数据, 去重, 并创建和提交事件
        MysqlUtil checkpointMysql = null;
        try {
            checkpointMysql = new MysqlUtil(this.checkPointUrl, false);
            List<JSONObject> filterResult = checkpointMysql.select(filterSql);
            this.sourceCounter.addToEventReceivedCount(filterResult.size());

            List<Event> events = new LinkedList<>();
            for (JSONObject jsonObject : filterResult) {
                String id = jsonObject.getString("id");
                String period_flag = jsonObject.getString("period_flag");
                String end_time = jsonObject.getString("end_time");

                // 创建事件
                SimpleEvent event = new SimpleEvent();
                // 给事件设置内容
                event.setBody(jsonObject.toJSONString().getBytes());
                String insertSql = String.format("insert into ma_checkpoint.tsr_marketing_activity_checkpoint(id,end_time,period_flag) value('%s','%s','%s') ON DUPLICATE KEY UPDATE checkpoint_status=1", id, end_time, period_flag);
                logger.info(insertSql);
                checkpointMysql.execute(insertSql);
                events.add(event);
            }
            // 提交数据
            getChannelProcessor().processEventBatch(events);
            checkpointMysql.commit();
            this.sourceCounter.addToEventAcceptedCount(events.size());
            // 删除过期数据
            deleteExpiredActive();
            Thread.sleep(this.pollIntervalSecond * 1000);
        } catch (SQLException e) {
            logger.error("get checkpoint mysql data has error: ", e);
            return Status.BACKOFF;
        } catch (Exception e) {
            logger.error("process event has error: ", e);
            return Status.BACKOFF;
        } finally {
            if (checkpointMysql != null)
                checkpointMysql.close();
        }
        return Status.READY;
    }

    /**
     * 删除checkpoint数据库里面已经过期的活动信息, 按照end_time来删除, 每天执行一次
     */
    public void deleteExpiredActive() {
        MysqlUtil checkpointMysql = null;
        try {
            String current = new SimpleDateFormat("yyyy-MM-dd 23:59:59").format(new java.util.Date(new Date().getTime() - this.deleteExpiredActiveIntervalDay * 24 * 60 * 60 * 1000));
            if (!this.deleteExpiredActiveFlag.equals(current)) {
                checkpointMysql = new MysqlUtil(this.checkPointUrl, false);
                String sql = String.format("delete from ma_checkpoint.tsr_marketing_activity_checkpoint where end_time = '%s'", current);
                logger.info(sql);
                checkpointMysql.execute(sql);
                checkpointMysql.commit();
                this.deleteExpiredActiveFlag = current;
            }
        } catch (Exception e) {
            if (checkpointMysql != null)
                checkpointMysql.rollback();
            logger.error("delete expired active data has error: ", e);
        } finally {
            if (checkpointMysql != null)
                checkpointMysql.close();
        }
    }


    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }

    @Override
    public synchronized void stop() {
        super.stop();
        this.sourceCounter.stop();
    }

}
